package com.markhsiu.common.thread;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 自定义线程池
 * @author Mark Hsiu
 *
 */
public class ThreadPoolTaskExecutor {

	private static final int workerThreads = Runtime.getRuntime().availableProcessors();
	private final Object poolSizeMonitor = new Object();
	// 线程池维护线程的最少数量
	private int corePoolSize = workerThreads;
	// 线程池维护线程的最大数量
	private int maxPoolSize = workerThreads * 100;
	// 线程池维护线程所允许的空闲时间
	private int keepAliveSeconds = 60;
	// 线程池所使用的缓冲队列大小
	private int queueCapacity = Integer.MAX_VALUE;

	private boolean allowCoreThreadTimeOut = false;
	private ThreadPoolExecutor threadPoolExecutor;

	ExecutorService initializeExecutor(ThreadFactory threadFactory, RejectedExecutionHandler rejectedExecutionHandler) {

		BlockingQueue<Runnable> queue = createQueue(this.queueCapacity);

		ThreadPoolExecutor executor = new ThreadPoolExecutor(this.corePoolSize, this.maxPoolSize, this.keepAliveSeconds,
				TimeUnit.SECONDS, queue, threadFactory, rejectedExecutionHandler);

		if (this.allowCoreThreadTimeOut) {
			executor.allowCoreThreadTimeOut(true);
		}

		this.threadPoolExecutor = executor;
		return executor;
	}

	/**
	 * Default is 1.
	 */
	public void setCorePoolSize(int corePoolSize) {
		synchronized (this.poolSizeMonitor) {
			this.corePoolSize = corePoolSize;
			if (this.threadPoolExecutor != null) {
				this.threadPoolExecutor.setCorePoolSize(corePoolSize);
			}
		}
	}

	/**
	 *
	 * Default is Integer.MAX_VALUE.
	 */
	public void setMaxPoolSize(int maxPoolSize) {
		synchronized (this.poolSizeMonitor) {
			this.maxPoolSize = maxPoolSize;
			if (this.threadPoolExecutor != null) {
				this.threadPoolExecutor.setMaximumPoolSize(maxPoolSize);
			}
		}
	}

	/**
	 * 
	 * Default is 60s.
	 */
	public void setKeepAliveSeconds(int keepAliveSeconds) {
		synchronized (this.poolSizeMonitor) {
			this.keepAliveSeconds = keepAliveSeconds;
			if (this.threadPoolExecutor != null) {
				this.threadPoolExecutor.setKeepAliveTime(keepAliveSeconds, TimeUnit.SECONDS);
			}
		}
	}

	public void setQueueCapacity(int queueCapacity) {
		this.queueCapacity = queueCapacity;
	}

	public void setAllowCoreThreadTimeOut(boolean allowCoreThreadTimeOut) {
		this.allowCoreThreadTimeOut = allowCoreThreadTimeOut;
	}

	protected BlockingQueue<Runnable> createQueue(int queueCapacity) {
		if (queueCapacity > 0) {
			return new LinkedBlockingQueue<Runnable>(queueCapacity);
		} else {
			return new SynchronousQueue<Runnable>();
		}
	}

	public ThreadPoolExecutor getThreadPoolExecutor() {
		if(this.threadPoolExecutor == null){
			throw new IllegalArgumentException("ThreadPoolTaskExecutor not initialized");
		}
		return this.threadPoolExecutor;
	}

	public int getActiveCount() {
		if (this.threadPoolExecutor == null) {
			return 0;
		}
		return this.threadPoolExecutor.getActiveCount();
	}

	public void execute(Runnable task) {
		Executor executor = getThreadPoolExecutor();
		try {
			executor.execute(task);
		} catch (RejectedExecutionException ex) {
			throw new RejectedExecutionException("Executor [" + executor + "] did not accept task: " + task, ex);
		}
	}

	public Future<?> submit(Runnable task) {
		ExecutorService executor = getThreadPoolExecutor();
		try {
			return executor.submit(task);
		} catch (RejectedExecutionException ex) {
			throw new RejectedExecutionException("Executor [" + executor + "] did not accept task: " + task, ex);
		}
	}

	public <T> Future<T> submit(Callable<T> task) {
		ExecutorService executor = getThreadPoolExecutor();
		try {
			return executor.submit(task);
		} catch (RejectedExecutionException ex) {
			throw new RejectedExecutionException("Executor [" + executor + "] did not accept task: " + task, ex);
		}
	}

}
